package com.atguigu.outputformat;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.eclipse.jetty.util.IO;

import java.io.IOException;

/*
    自定义RecordWriter
 */
public class MyRecordWriter extends RecordWriter<LongWritable, Text> {

    private FSDataOutputStream atguigu;
    private  FSDataOutputStream other;
    /*
        静态代码块
        非静态代码块
        构造器
     */
    //创建流
    public MyRecordWriter(TaskAttemptContext job) throws IOException {
        //1.创建文件系统对象
        FileSystem fs = FileSystem.get(job.getConfiguration());
        //2.创建流
        //2.1获取输出路径
        Path outputPath = FileOutputFormat.getOutputPath(job);
        //2.2创建文件路径
        Path atguiguPath = new Path(outputPath, "atguigu.txt");//D:\io\output33333\atguigu.txt
        Path otherPath = new Path(outputPath, "other.txt");
        //2.3创建流
        atguigu = fs.create(atguiguPath);
        other = fs.create(otherPath);

    }

    /**
     * 写数据
     * 该被方法在被循环调用
     * @param key 偏移量
     * @param value 一行一行的内容
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public void write(LongWritable key, Text value) throws IOException, InterruptedException {
        String address = value.toString() + "\n";
        //判断内容中是否包含atguigu
        if (address.contains("atguigu")){
            //写到atguigu.txt
            atguigu.write(address.getBytes());
        }else{
            //写到other.txt
            other.write(address.getBytes());
        }
    }

    /*
        在该方法中用来关闭资源
     */
    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        //关资源
        IOUtils.closeStream(atguigu);
        IOUtils.closeStream(other);
    }
}
